package org.apache.rocketmq.client.java.impl;

import apache.rocketmq.v2.*;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import io.grpc.Metadata;
import io.grpc.stub.StreamObserver;
import org.apache.rocketmq.client.java.clientapis.ClientException;
import org.apache.rocketmq.client.java.misc.ExecutorServices;
import org.apache.rocketmq.client.java.misc.ThreadFactoryImpl;
import org.apache.rocketmq.client.java.route.Endpoints;
import org.apache.rocketmq.client.java.rpc.RpcClient;
import org.apache.rocketmq.client.java.rpc.RpcClientImpl;
import org.apache.rocketmq.client.java.rpc.RpcInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.SSLException;
import java.io.IOException;
import java.time.Duration;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ClientManagerImpl extends ClientManager {
    //rpc客户端的最大空闲时间，如果超过这个时间，客户端没有和服务端进行通信
    //那就会把这个rpc客户端认定为空闲客户端，程序内部会定期检查是否存在空闲客户端，然后释放对应的空闲客户端资源
    public static final Duration RPC_CLIENT_MAX_IDLE_DURATION = Duration.ofMinutes(30);

    //检查空闲客户端的定时任务的初始延迟时间
    public static final Duration RPC_CLIENT_IDLE_CHECK_INITIAL_DELAY = Duration.ofSeconds(5);

    //检查空闲客户端的定时任务的执行周期
    public static final Duration RPC_CLIENT_IDLE_CHECK_PERIOD = Duration.ofMinutes(1);

    //定时发送心跳请求的定时任务的初始延迟时间
    public static final Duration HEART_BEAT_INITIAL_DELAY = Duration.ofSeconds(1);

    //定时发送心跳请求的定时任务的执行周期
    public static final Duration HEART_BEAT_PERIOD = Duration.ofSeconds(10);

    //定时记录日志的定时任务的初始延迟时间
    public static final Duration LOG_STATS_INITIAL_DELAY = Duration.ofSeconds(60);

    //定时记录日志的定时任务的执行周期
    public static final Duration LOG_STATS_PERIOD = Duration.ofSeconds(60);

    //定时发送同步设置请求的定时任务的初始延迟时间
    public static final Duration SYNC_SETTINGS_DELAY = Duration.ofSeconds(1);

    //定时发送同步设置请求的定时任务的执行周期
    public static final Duration SYNC_SETTINGS_PERIOD = Duration.ofMinutes(5);

    private static final Logger LOGGER = LoggerFactory.getLogger(ClientManagerImpl.class);

    //客户端对象
    private final Client client;

    //RPC客户端连接表，key为服务端地址，value为对应的RPC客户端连接
    @GuardedBy("rpcClientTableLock")
    private final Map<Endpoints, RpcClient> rpcClientTable;

    //保护客户端连接表并发安全读写锁
    private final ReadWriteLock rpcClientTableLock ;

    //定时任务执行器
    private final ScheduledExecutorService scheduler;

    //异步工作执行器，执行器用于运行异步任务，比如发送心跳请求，同步设置请求，检查空闲客户端，记录日志等等
    private final ExecutorService asyncWorker;

    public ClientManagerImpl(Client client){
        this.client = client;
        this.rpcClientTable = new ConcurrentHashMap<>();
        this.rpcClientTableLock = new ReentrantReadWriteLock();
        this.scheduler = new ScheduledThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors(),
                new ThreadFactoryImpl("ClientScheduler"));

        this.asyncWorker = new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors(),
                Runtime.getRuntime().availableProcessors(),
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                new ThreadFactoryImpl("ClientAsyncWorker"));
    }
    @Override
    public ScheduledExecutorService getScheduler()  {
        return this.scheduler;
    }
    /*** @方法描述：清理空闲的rpc客户端连接的方法
     */
    private void clearIdleRpcClients() throws InterruptedException {
        //获取写锁，保证并发安全
        rpcClientTableLock.writeLock().lock();
        try {
            for (Iterator<Map.Entry<Endpoints, RpcClient>> iterator = rpcClientTable.entrySet().iterator(); iterator.hasNext(); ) {
                Map.Entry<Endpoints, RpcClient> entry = iterator.next();
                //得到服务端地址
                Endpoints endpoints = entry.getKey();
                //得到对应的RPC客户端连接对象
                RpcClient rpcClient = entry.getValue();
                //判断RPC客户端连接对象是否空闲
                Duration idleDuration = rpcClient.idleDuration();
                if (idleDuration.compareTo(RPC_CLIENT_MAX_IDLE_DURATION) > 0) {
                    //从客户端连接表中删除该RPC客户端连接对象
                    iterator.remove();
                    //如果RPC客户端连接对象已经空闲超过指定的时间，就关闭该RPC客户端连接对象
                    rpcClient.shutdown();
                    LOGGER.info("Rpc client has been idle for a long time, endpoints={}, idleDuration={}, " +
                            "rpcClientMaxIdleDuration={}", endpoints, idleDuration, RPC_CLIENT_MAX_IDLE_DURATION);
                }

            }
        }finally {
            rpcClientTableLock.writeLock().unlock();
        }
    }
    /**
     * @方法描述：得到指定服务端地址对应的RPC客户端连接对象
     */
    private RpcClient getRpcClient(Endpoints endpoints) throws ClientException {
        RpcClient rpcClient;
        //得到读锁，保证并发安全
        rpcClientTableLock.readLock().lock();
        try {
            //先从客户端连接表中查找该服务端地址对应的RPC客户端连接对象
            rpcClient = rpcClientTable.get(endpoints);
            //如果找到了，就直接返回该RPC客户端连接对象
            if (null != rpcClient) {
                return rpcClient;
            }
        } finally {
            //释放读锁
            rpcClientTableLock.readLock().unlock();
        }
        //如果没有从客户端连接表中找到对应的rpc客户端对象
        //那就在这里获取写锁，因为接下来要为这个服务端地址创建一个对应的rpc客户端连接对象
        //然后把该rpc客户端对象放入客户端连接表中，获取写锁保证并发安全
        rpcClientTableLock.writeLock().lock();
        try {
            //双重判断，因为可能在获取写锁之前，其他线程已经创建了该服务端地址对应的RPC客户端连接对象
            rpcClient = rpcClientTable.get(endpoints);
            if (null != rpcClient) {
                return rpcClient;
            }
            try {
                //仍然没有找到对应的rpc客户端对象，那就在这里创建一个对应的rpc客户端对象
                rpcClient = new RpcClientImpl(endpoints);
            } catch (SSLException e) {
                LOGGER.error("Failed to get rpc client, endpoints={}", endpoints);
                throw new ClientException("Failed to generate RPC client", e);
            }
            //把新创建的rpc客户端对象放入客户端连接表中
            rpcClientTable.put(endpoints, rpcClient);
            return rpcClient;
        } finally {
            //释放写锁
            rpcClientTableLock.writeLock().unlock();
        }
    }
    @Override
    public ListenableFuture<RpcInvocation<QueryRouteResponse>> queryRoute(Endpoints endpoints, Metadata metadata, QueryRouteRequest request, Duration duration) {

        try {
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.queryRoute(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送心跳请求的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<HeartbeatResponse>> heartbeat(Endpoints endpoints, Metadata metadata,
                                                                        HeartbeatRequest request, Duration duration) {
        try {
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.heartbeat(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送消息的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<SendMessageResponse>> sendMessage(Endpoints endpoints, Metadata metadata,
                                                                            SendMessageRequest request, Duration duration) {
        try {
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.sendMessage(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端查询分配的队列信息的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<QueryAssignmentResponse>> queryAssignment(Endpoints endpoints,
                                                                                    Metadata metadata, QueryAssignmentRequest request, Duration duration) {
        try {
            //得到指定服务端地址对应的RPC客户端连接对象
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.queryAssignment(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送接收消息的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<Iterator<ReceiveMessageResponse>>> receiveMessage(Endpoints endpoints,
                                                                                            Metadata metadata, ReceiveMessageRequest request, Duration duration) {
        try {
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.receiveMessage(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送确认消息的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<AckMessageResponse>> ackMessage(Endpoints endpoints, Metadata metadata,
                                                                          AckMessageRequest request, Duration duration) {
        try {
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.ackMessage(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送修改消息不可见时间请求的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<ChangeInvisibleDurationResponse>> changeInvisibleDuration(
            Endpoints endpoints, Metadata metadata, ChangeInvisibleDurationRequest request, Duration duration) {
        try {
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.changeInvisibleDuration(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送转发消息到死信队列的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<ForwardMessageToDeadLetterQueueResponse>> forwardMessageToDeadLetterQueue(
            Endpoints endpoints, Metadata metadata, ForwardMessageToDeadLetterQueueRequest request, Duration duration) {
        try {
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.forwardMessageToDeadLetterQueue(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送结束事务请求的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<EndTransactionResponse>> endTransaction(Endpoints endpoints,
                                                                                  Metadata metadata, EndTransactionRequest request, Duration duration) {
        try {
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.endTransaction(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送客户端即将终止的请求的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<NotifyClientTerminationResponse>> notifyClientTermination(
            Endpoints endpoints, Metadata metadata, NotifyClientTerminationRequest request, Duration duration) {
        try {
            final RpcClient rpcClient = getRpcClient(endpoints);
            return rpcClient.notifyClientTermination(metadata, request, asyncWorker, duration);
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送遥测请求的方法
     */
    @Override
    public StreamObserver<TelemetryCommand> telemetry(Endpoints endpoints, Metadata metadata, Duration duration,
                                                      StreamObserver<TelemetryCommand> responseObserver) throws ClientException {
        final RpcClient rpcClient = getRpcClient(endpoints);
        return rpcClient.telemetry(metadata, asyncWorker, duration, responseObserver);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：启动客户端管理器的方法
     */
    @Override
    protected void startUp() {

        LOGGER.info("Begin to start the client manager");
        //向定时任务执行器提交空闲rpc客户端清理任务
        scheduler.scheduleWithFixedDelay(
                () -> {
                    try {
                        clearIdleRpcClients();
                    } catch (Throwable t) {
                        LOGGER.error("Exception raised while clear idle rpc clients.", t);
                    }
                },
                RPC_CLIENT_IDLE_CHECK_INITIAL_DELAY.toNanos(),
                RPC_CLIENT_IDLE_CHECK_PERIOD.toNanos(),
                TimeUnit.NANOSECONDS
        );

        //向定时任务执行器提交心跳任务
        scheduler.scheduleWithFixedDelay(
                () -> {
                    try {
                        client.doHeartbeat();
                    } catch (Throwable t) {
                        LOGGER.error("Exception raised while heartbeat.", t);
                    }
                },
                HEART_BEAT_INITIAL_DELAY.toNanos(),
                HEART_BEAT_PERIOD.toNanos(),
                TimeUnit.NANOSECONDS
        );

        //向定时任务执行器提交日志统计任务
        scheduler.scheduleWithFixedDelay(
                () -> {
                    try {
                        client.doStats();
                    } catch (Throwable t) {

                        LOGGER.error("Exception raised while log stats.", t);
                    }
                },
                LOG_STATS_INITIAL_DELAY.toNanos(),
                LOG_STATS_PERIOD.toNanos(),
                TimeUnit.NANOSECONDS
        );

        //向定时任务执行器提交客户端向服务端同步设置任务
        scheduler.scheduleWithFixedDelay(
                () -> {
                    try {
                        client.syncSettings();
                    } catch (Throwable t) {
                        LOGGER.error("Exception raised during the setting synchronizing.", t);
                    }
                },
                SYNC_SETTINGS_DELAY.toNanos(),
                SYNC_SETTINGS_PERIOD.toNanos(),
                TimeUnit.NANOSECONDS
        );
        LOGGER.info("The client manager starts successfully");
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：终止客户端管理器工作的方法
     */
    @Override
    protected void shutDown() throws IOException {
        LOGGER.info("Begin to shutdown the client manager");
        //先终止定时任务执行器工作，停止接收新任务，但允许已提交的任务继续执行完成
        scheduler.shutdown();
        try {
            //等待定时任务执行器中的任务完成或超时
            if (!ExecutorServices.awaitTerminated(scheduler)) {
                //等待超时则记录错误日志
                LOGGER.error("[Bug] Timeout to shutdown the client scheduler");
            } else {
                //执行器正常关闭则记录成功日志
                LOGGER.info("Shutdown the client scheduler successfully");
            }

            //获取客户端连接表的写锁，接下来开始释放rpc客户端连接资源
            rpcClientTableLock.writeLock().lock();
            try {
                //下面的操作都很简单，我就不添加注释了，大家自己看看就行
                final Iterator<Map.Entry<Endpoints, RpcClient>> it = rpcClientTable.entrySet().iterator();
                while (it.hasNext()) {
                    final Map.Entry<Endpoints, RpcClient> entry = it.next();
                    final RpcClient rpcClient = entry.getValue();
                    it.remove();
                    rpcClient.shutdown();
                }
            } finally {
                //释放写锁
                rpcClientTableLock.writeLock().unlock();
            }

            //记录所有rpc客户端成功关闭的日志信息
            LOGGER.info("Shutdown all rpc client(s) successfully");

            //关闭异步工作执行器，停止接收新任务，但允许已提交的任务继续执行完成
            asyncWorker.shutdown();
            //等待异步工作执行器中的任务完成或超时
            if (!ExecutorServices.awaitTerminated(asyncWorker)) {
                LOGGER.error("[Bug] Timeout to shutdown the client async worker");
            } else {
                LOGGER.info("Shutdown the client async worker successfully");
            }
        } catch (InterruptedException e) {
            LOGGER.error("[Bug] Unexpected exception raised while shutdown client manager", e);
            throw new IOException(e);
        }
        LOGGER.info("Shutdown the client manager successfully");
    }
}